package start.demo01.datatype.datasource;

import lombok.Data;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

@Data
public class KafkaPool {

    public KafkaPool() {
        //"192.168.150.151:9092"
    }

    public void send(Producer producer,String topic,String msg) {
        try {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, msg);
            producer.send(record);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
    public Producer producer(String servers) {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", servers);
        properties.put("acks", "all");
        properties.put("retries", 0);
        properties.put("batch.size", 16384);
        properties.put("linger.ms",1);//等待时间
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer;
        producer = new KafkaProducer(properties);
        return producer;
    }

}
